package com.cskaoyan.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @auther cskaoyan
 * @date 2022/7/19:10:53
 */
@Component
public class CreatePromoOrderProducer {


    private TransactionMQProducer producer;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");

        producer.setNamesrvAddr("127.0.0.1:9876");

        // 设置事物监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                Map<String, Long> parmMap = (Map<String, Long>) arg;

                // 本地事务，扣减秒杀商品库存
                Integer effectiveRow = 0;
                Long productId = parmMap.get("productId");
                Long psId = parmMap.get("psId");

                // 加分布式锁
                // 1. 获取锁
                String lockKey = "promo_item: " + productId + "-" + psId;
                RLock lockObj = redissonClient.getLock(lockKey);


                // 先加锁, 加锁失败的线程会阻塞等待
                lockObj.lock();
                try {
                    effectiveRow = promoItemMapper.decreaseStock(productId, psId);
                } finally {
                    // 保证锁一定会被释放
                    lockObj.unlock();
                }



                String localTransactionKey = "promo_local_transaction:" + msg.getTransactionId();
                if (effectiveRow < 1) {
                    // 秒杀商品售罄(秒杀下单应该失败)，不需要让下游服务消费到事物消息

                    // 针对售罄的商品，在redis中打库存售罄的标志
                    String notEngouhKey = "promo_item_stock_not_enough: " + productId + "-" + psId;
                    cacheManager.setCache(notEngouhKey, "none", 1);

                    cacheManager.setCache(localTransactionKey, "ROLLBACK_MESSAGE", 1);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                // 库存扣减成功
                cacheManager.setCache(localTransactionKey, "COMMIT_MESSAGE", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 构造获取本地事务执行结果的key值
                String localTransactionKey = "promo_local_transaction:" + msg.getTransactionId();

                String s = cacheManager.checkCache(localTransactionKey);
                if (s == null || s.isEmpty()) {
                    // 还没有该本地事务对应的结果
                    return LocalTransactionState.UNKNOW;
                }

                // 本地事务执行失败
                if ("ROLLBACK_MESSAGE".equals(s)) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                // 本地事物执行成功
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        try {
            // 启动
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }




    public boolean sendPromoOrderTransactionMessage(CreateSeckillOrderRequest request, Long productId, Long psId) {

        // 先将参数转化为Json字符串
        String paramJson = JSON.toJSONString(request);
        byte[] paramJsonBytes = paramJson.getBytes(Charset.forName("utf-8"));

        HashMap<String, Long> paramMap = new HashMap<>();
        paramMap.put("psId", psId);
        paramMap.put("productId", productId);

        // 准备事物消息，并封装数据
        Message message = new Message("create_promo_order", paramJsonBytes);


        //发送事物消息
        TransactionSendResult sendResult = null;
        try {
            // sendMessageInTransaction的第二个参数，是传递给executeLocalTransaction方法使用的参数
            // 即第二个参数，代表的是本地事务执行所需要的参数
          sendResult = producer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        //
        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
            // 本地事务执行成功(事物消息发送成功)
            return true;
        }

        // 事物消息发送失败 or 本地事务执行失败
        return false;
    }
}
